/**
 * Wihidum - Efficient Broker Hierarchy for Publish Subscribe Middleware
 *
 * License      :   Apache License 2.0
 * Home Page    :   http://code.google.com/p/wihidum/
 * Mailing List :   wihidum@googlegroups.com
 */

package lk.mrt.cse.wihidum.receiver;

import lk.mrt.cse.wihidum.delivery.*;
import lk.mrt.cse.wihidum.processor.*;
import lk.mrt.cse.wihidum.util.*;
import lk.mrt.cse.wihidum.storage.*;
import org.apache.axiom.om.*;
import org.apache.axiom.soap.*;
import org.apache.axis2.*;
import org.apache.axis2.context.*;
import org.apache.axis2.receivers.*;
import org.apache.log4j.*;
import org.apache.axis2.client.*;

import javax.xml.namespace.QName;
import java.util.UUID;
import java.util.HashMap;


/**
 * Entry point of the messages received to the service
 */
public class MessageReceiver
        extends AbstractInOutMessageReceiver
{

    private static final Logger logger = Logger.getLogger (MessageReceiver.class);
    private static ThreadPool threadPool = new ThreadPool ();
    private Processor publicationProcessor;
    private SubscriptionProcessor subscriptionProcessor;
    private Processor adminProcessor;

    /**
     * Default contructor
     */
    public MessageReceiver ()
    {
        publicationProcessor = new PublicationProcessor ();
        subscriptionProcessor = new SubscriptionProcessor ();
        adminProcessor =  new AdminProcessor();
    }

    /**
     * Process the messages
     *
     * @param inMessage  Incoming message context
     * @param outMessage Outgoing message context
     * @throws AxisFault Exception
     */
    public void invokeBusinessLogic (MessageContext inMessage, MessageContext outMessage)
            throws AxisFault
    {
        //logger.info ("Broker service received a new message");

        try
        {
            Options opts=inMessage.getOptions();
            String action=opts.getAction();
            // Get hold of the request SOAP Envelope
            SOAPEnvelope requestEnvelop = inMessage.getEnvelope ();

            // The response element
            OMElement responseElement;
            SOAPEnvelope responseEnvelope;

            if (action.equals (WihidumConstants.Operations.SUBSCRIBE))
            {
                String messageID=createMessageID();
                logger.info ("Received message to Broker Service is a subscription request#"+messageID);

                // Hand over the message to the subscription processor
                responseEnvelope = (SOAPEnvelope) subscriptionProcessor.processMessage(requestEnvelop, messageID);

                Options outOpts=outMessage.getOptions();
                outOpts.setAction(WihidumConstants.Actions.SubscribeResponse);
                // Set the response envelope in the response message context
                outMessage.setEnvelope (responseEnvelope);
                logger.info("Subscription response sent to message#"+messageID);
            }
            else if (action.equals(WihidumConstants.Operations.BROKER_SUBSCRIBE))
            {
                String messageID=decodeMessageID(requestEnvelop);
                logger.info ("Received message to Broker Service is a broker subscription request#"+messageID);

                // Hand over the message to the subscription processor
                responseEnvelope = (SOAPEnvelope) subscriptionProcessor.processPropogatingMessage (requestEnvelop,messageID);

                // Set the response envelope in the response message context
                Options outOpts=outMessage.getOptions();
                outOpts.setAction(WihidumConstants.Operations.BROKER_SUBSCRIBE_RESPONSE);

                outMessage.setEnvelope (responseEnvelope);

                //logger.info("Broker subscription response sent");
            }
            else if (action.equals(WihidumConstants.Operations.BROKER_PUBLISH))
            {
                String randomID="";
                logger.info ("Received message to Broker Service is a broker publication request");

                // Hand over the message to the publication processor
                responseElement = publicationProcessor.processPropogatingMessage (requestEnvelop,randomID);

                // Set the response envelope in the response message context
                Options outOpts=outMessage.getOptions();
                outOpts.setAction("brokerPublishResponse");
                outMessage.setEnvelope (createResponse (responseElement));
            }
            else if(action.equals(WihidumConstants.Actions.Unsubscribe))
            {
                logger.info ("Received message to Broker Service is an Unsubscription request");                
                 SOAPEnvelope unsubscriptionResponse=subscriptionProcessor.processUnsubscribeMessage(requestEnvelop);
                 Options outOpts=outMessage.getOptions();
                 outOpts.setAction(WihidumConstants.Actions.UnsubscribeResponse);
                 outMessage.setEnvelope(unsubscriptionResponse);
            }
            else if (action.equals (WihidumConstants.Operations.CHANGE_TOPOLOGY))
            {
                logger.info ("Received message to Broker Service is a Change Topology request");
                responseEnvelope = adminProcessor.processMessage(requestEnvelop, "");
                Options outOpts=outMessage.getOptions();
                outOpts.setAction(WihidumConstants.Operations.CHANGE_TOPOLOGY_RESPONSE);
                outMessage.setEnvelope (responseEnvelope);
                logger.info ("Received message processing complete");
            }
            else if(action.equals(WihidumConstants.Operations.BROKER_UNSUBSCRIBE))
            {
                logger.info ("Received message to Broker Service is a broker unsubscription");
                responseEnvelope=subscriptionProcessor.processBrokerUnsubscribeMessage(requestEnvelop);
                Options outOpts=outMessage.getOptions();
                outOpts.setAction(WihidumConstants.Operations.BROKER_UNSUBSCRIBE_RESPONSE);
                outMessage.setEnvelope (responseEnvelope);
            }
        }
        catch (NullPointerException ex)
        {
            logger.error ("Error in recognizing received message format", ex);
        }
        catch (OMException ex)
        {
            logger.error ("Error in recognizing received message format", ex);
        }
        catch (AxisFault ex)
        {
            logger.error ("Error in recognizing received message format", ex);
        }
    }
    private String createMessageID()
    {
        String logUUID= UUID.randomUUID().toString();
        String myID=SelfReader.getMyID();
        String messageID=logUUID+myID;
        return messageID;
    }
    private String decodeMessageID(SOAPEnvelope messageEnvelope)
    {
        SOAPBody requestBody=messageEnvelope.getBody();
        OMElement logIDElement=requestBody
                        .getFirstChildWithName(new QName(WSEventingConstants.WIHIDUM_NAMESPACE,"BrokerSubscription"))
                        .getFirstChildWithName(new QName(WSEventingConstants.WIHIDUM_NAMESPACE,"LogID"));
        String logID=logIDElement.getText();
        return logID;
    }

    /**
     * Create response SOAP envelop
     *
     * @param element Response element
     * @return Created response envelop
     */
    private SOAPEnvelope createResponse (OMElement element)
    {
        // Create the response SOAP facotry
        SOAPFactory factory = OMAbstractFactory.getSOAP11Factory ();

        // Create the response SOAP envelop
        SOAPEnvelope responseEnvelop = factory.getDefaultEnvelope ();

        // Add the response to the response SOAP envelop
        responseEnvelop.getBody ().addChild (element);

        return responseEnvelop;
    }

    /**
     * ThreadPool getter method
     *
     * @return ThreadPool
     */
    public static ThreadPool getThreadPool ()
    {
        return threadPool;
    }

    /**
     * Publication processor setter method
     *
     * @param publicationProcessor PublicationProcessor
     */
    public void setPublicationProcessor (Processor publicationProcessor)
    {
        this.publicationProcessor = publicationProcessor;
    }

    /**
     * Subscription processor setter method
     *
     * @param subscriptionProcessor SubscriptionProcessor
     */
    public void setSubscriptionProcessor (Processor subscriptionProcessor)
    {
       // this.subscriptionProcessor = subscriptionProcessor;
    }

}
